都会|可能会_C#RabbitMQ入门指南

篇首语：本文由编程笔记#小编为大家整理，主要介绍了C#RabbitMQ入门指南相关的知识，希望对你有一定的参考价值。

文章目录

• 1.简介
• 2. 相关概念
• • 2.1 消息中间件
  • 2.2 消息中间件的作用
  • 2.3 RabbitMQ中的一些概念
  • 2.4 RabbitMQ模型
  
  
• 3. ※点对点模型
• • 3.1 轮询消费&＃xff08;自动ack&＃xff09;
  • 3.2 ※手动发送ACK与数据持久存储
  • 3.3 消费模式
  
  
• 4. ※发布订阅模型
• • 4.1 交换机&＃xff08;Exchange&＃xff09;
  • 4.2 &＃96;fanout&＃96;交换机
  • 4.3 路由&＃xff08;Routing&＃xff09;
  • • 4.3.1 &＃96;Direct&＃96;交换机
    
    
  • 4.4 主题&＃xff08;Topics&＃xff09;
  • • 4.4.1 &＃96;Topic&＃96;交换机
    • 4.4.2 最终实现
    
    
  
  
• 5. 实现远程过程调用&＃xff08;RPC&＃xff09;

RabbitMQ 是采用 erlang 语言实现 AMQP (Advanced Message Queuing Protocol &＃xff0c;高级消息
队列协议)的消息中间件&＃xff0c;它最初起源于金融系统&＃xff0c;用于在分布式系统中存储转发消息.

RabbitMQ 是目前非常热门的一款消息中间件&＃xff0c;不管是互联网行业还是传统行业都在大量
地使用 RabbitMQ 凭借其高可靠、易扩展、高可用及丰富的功能特性受到越来越多企业的青睐。

RabbitMQ的具体特点可以概括为以下几点。

• 可靠性&＃xff1a;RabbitMQ使用一些机制来保证可靠性。如持久化、传输确认及发布确认等。
• 灵活的路由&＃xff1a;在消息进入队列之前&＃xff0c;通过交换机来路由消息。对于典型的路由功能&＃xff0c;提供了一些内置的交换机来实现。针对更复杂的路由功能&＃xff0c;可以将多个交换机绑定在一起&＃xff0c;可以通过插件机制来实现自己的交换机。
• 扩展性&＃xff1a;多个MQ节点可以组成一个集群&＃xff0c;也可以根据实际业务情况动态地扩展
  集群中节点。
• 高可用性&＃xff1a;队列可以在集群中的机器上设置镜像&＃xff0c;使得在部分节点出现问题的情况下队
  列仍然可用。
• 多种协议&＃xff1a;除了原生支持AMQP协议&＃xff0c;还支持STOMP、MQTT等多种消息
  中间件协议。
• 多语言客户端:几乎支持所有常用语言&＃xff0c;比如C#、Java、Python、Ruby、php、Javascript等。
• 管理界面&＃xff1a;RabbitMQ 提供了一个易用的用户界面&＃xff0c;使得用户可以监控和管理消息、集
  群中的节点等。
• 插件机制: RabbitMQ 提供了许多插件 以实现从多方面进行扩展&＃xff0c;当然也可以编写自
  己的插件。

依赖

• 本文基于发稿时RabbitMQ的最新版本&＃xff1a;3.8.19.

• RabbitMQ客户端使用&＃xff1a;RabbitMQ.Client 6.2.2

• RabbitMQ可视化管理插件安装&＃xff1a;官网。首先执行rabbitmq-plugins enable rabbitmq_management命令&＃xff0c;然后打开管理面板&＃xff1a;http://localhost:15672/#/ 即可&＃xff0c;默认用户名密码都是guest。

2.1 消息中间件

消息 (Message)&＃xff1a;是指在应用间传送的数据。消息可以非常简单&＃xff0c;比如只包含文本字符串、JSON 等&＃xff0c;也可以很复杂&＃xff0c;比如内嵌对象。

消息队列中间件 (Message Queue Middleware&＃xff0c;简称为 MQ) 是指利用高效可靠的消息传递机制进行与平台无关的数据交流&＃xff0c;并基于数据通信来进行分布式系统的集成。通过提供消息传和消息排队模型&＃xff0c;它可以在分布式环境下扩展进程间的通信。它一般有两种传递模式:点对点(P2P, Point-to-Point) 模式和发布/订阅 (Pub/Sub) 模式。

点对点模式是基于队列的&＃xff0c;消息生产发送消息到队列&＃xff0c;消息消费者从队列中接收消息&＃xff0c;队列的存在使得消息的异步传输成为可能。

发布订阅模式定义了如何向一个内容节点发布和订阅消息&＃xff0c;这个内容节点称为主题 (topic) &＃xff0c;主题可以认为是消息传递的中介&＃xff0c;消息发布者将消息发布到某个主题&＃xff0c;而消息订阅者则从主题中订阅消息。主题使得消息的订阅者与消息的发布者互相保持独立&＃xff0c;不需要进行接触即可保证消息的传递&＃xff0c;发布/订阅模式在消息的一对多广播时采用。

2.2 消息中间件的作用

• 解耦: 最大的作用其实是解耦。
• 冗余存储&＃xff1a;有些情况下&＃xff0c;处理数据的过程会失败。消息中间件可以把数据进行持久化直
  到它们已经被完全处理&＃xff0c;通过这一方式规避了数据丢失风险。在把一个消息从消息中间件中删除之前&＃xff0c;需要你的处理系统明确地指出该消息己经被处理完成&＃xff0c;从而确保你的数据被安全地保存直到你使用完毕。
• 扩展性: 因为消息中间件解耦了应用的处理过程&＃xff0c;所以提高消息入队和处理的效率是很容易的&＃xff0c;只要另外增加处理过程即可&＃xff0c;不需要改变代码&＃xff0c;也不需要调节参数。
• 削峰: 在访问量剧增的情况下&＃xff0c;应用仍然需要继续发挥作用&＃xff0c;但是这样的突发流 并不常
  见。如果以能处理这类峰值为标准而投入资源&＃xff0c;无疑是巨大的浪费。使用消息中间件能够使关键组件支撑突发访问压力&＃xff0c;不会因为突发的超负荷请求而完全崩惯
• 可恢复性: 当系统一部分组件失效时&＃xff0c;不会影响到整个系统。消息中间件降低了进程间的耦合度&＃xff0c;所以即使一个处理消息的进程挂掉&＃xff0c;加入消息中间件中的消息仍然可以在系统恢复后进行处理
• 顺序保证: 在大多数使用场景下&＃xff0c;数据处理的顺序很重要&＃xff0c;大部分消息中间件支持一定程度上的顺序性。
• 缓冲: 在任何重要的系统中&＃xff0c;都会存在需要不同处理时间的元素。消息中间件通过一个缓冲层来帮助任务最高效率地执行&＃xff0c;写入消息中间件的处理会尽可能快速。该缓冲层有助于控制和优化数据流经过系统的速度。
• 异步通信: 在很多时候应用不想也不需要立即处理消息 消息中间件提供了异步处理机制&＃xff0c;允许应用把 些消息放入消息中间件中&＃xff0c;但并不立即处理它&＃xff0c;在之后需要的时候再慢慢处理。

2.3 RabbitMQ中的一些概念

RabbitMQ的整体模型架构如下&＃xff1a;

1. Producer&＃xff1a;生产者&＃xff0c;用来生产消息。并把消息发给交换机&＃xff08;生产者不会把消息直接发给某个队列&＃xff0c;很多图你可能会看到生产者直连队列&＃xff0c;其实中间隐藏了一个默认的交换机&＃xff09;。生产者也就是发送消息的一方。

2. Consumer&＃xff1a;消费者&＃xff0c;用来消费队列里的消息。也就是接受消息的一方。

3. Exchange&＃xff1a;交换机&＃xff0c;有些文章会成为交换器。其实这个东西的作用更像是路由器。交换机会根据生产者发过来的消息的routingKey&＃xff0c;把消息丢到不同的队列中。

4. Queue&＃xff1a;队列&＃xff0c;用来存储交换机丢过来的消息&＃xff08;可以理解为邮箱&＃xff09;。一个队列可以被多个消费者进行消费&＃xff0c;此时队列里的消息会按照轮询的方式一个个的分配给下面的消费者&＃xff08;不支持队列层面的广播消费&＃xff09;。
   

5. channel: 通道&＃xff0c;RabbitMQ 处理的每条 AMQP 指令都是通过通道完成的。如下图所示。通道的存在其实就是为了复用TCP连接&＃xff0c;本质上我们也可以使用TCP连接发送命令。但是当应用中有多个线程需要生产或者消费时&＃xff0c;就需要创建多个TCP连接&＃xff0c;而TCP连接的创建和销毁很费资源。
   

6. routingKey&＃xff1a;路由键&＃xff0c;交换机根据这个的值来决定把消息丢到哪个队列里&＃xff0c;没有队列可以接受的话&＃xff0c;可能把消息返回给生产者也可能直接丢弃。

7. Broker&＃xff1a;RabbitMQ的服务节点或服务实例。可以简单里的理解为就是一台RabbitMQ服务器。

8. Binding&＃xff1a;绑定&＃xff0c;消费者端就行配置&＃xff0c;建立队列与某个交换机的关系&＃xff0c;这样交换机收到消息之后就知道是否要投递到这个队列了。

2.4 RabbitMQ模型

可以看到官网的教程里有六种模型&＃xff1a;

看起来很多很唬人&＃xff0c;但是不要怕&＃xff0c;本质上也就以下两种&＃xff0c;学起来也很快。

1. 点对点&＃xff1a;前两种就是属于点对点模型&＃xff0c;即队列里的一个消息只能被一个消费者消费。第二种是对第一种的扩充&＃xff0c;额外增加了一个消费者而已。多个消费者就是采用轮询的机制去消费同一个队列里的消息。
2. 发布订阅&＃xff1a;剩下的4种都是发布订阅模型&＃xff0c;即生产者发布的一个消息可以被N个消费者消费&＃xff0c;实现方式是通过交换机把同一个消息投递到了N个队列里。4、5、6都是对3的功能扩充&＃xff0c;让你有更大的自由度来决定一个消息能投递到哪个队列里。

队列的一个消息只能被一个消费者消费&＃xff0c;多个消费者可以通过轮询的方式消费也可以通过手动响应ack的方式竞争消费。

3.1 轮询消费&＃xff08;自动ack&＃xff09;

当开启一个消费者实例时模型如下&＃xff1a;

当开启两个消费者实例时模型如下&＃xff1a;

消费者示例&＃xff1a;
channel.BasicConsume的第二个参数为true&＃xff0c;表示当消费者收到消息后&＃xff08;非消息的业务逻辑处理完后&＃xff09;&＃xff0c;会自动发送一个ack给mq表示消息已收到。

static void Main(string[] args)

var factory &＃61; new ConnectionFactory() HostName &＃61; "localhost" ;
using (var connection &＃61; factory.CreateConnection())

using (var channel &＃61; connection.CreateModel())

channel.QueueDeclare("hello", false, false, false, null);
var consumer &＃61; new EventingBasicConsumer(channel);
consumer.Received &＃43;&＃61; (model, args) &＃61;>

byte[] body &＃61; args.Body.ToArray();
var msg &＃61; Encoding.UTF8.GetString(body);
Console.WriteLine(" [x] Received 0", msg);
;
//第二个参数autoAck为true
channel.BasicConsume("hello", true, consumer);
Console.WriteLine("Press [Enter] to exit");
Console.ReadKey();






生产者示例&＃xff1a;

static void Main(string[] args)

var factory &＃61; new ConnectionFactory() HostName &＃61; "localhost" ;
using (var connection &＃61; factory.CreateConnection())

using (var channel &＃61; connection.CreateModel())

//声明队列操作是幂等的&＃xff0c;当队列不存在时&＃xff0c;会进行创建。
channel.QueueDeclare(queue: "hello", durable: false, exclusive: false, autoDelete: false, arguments: null);
Console.WriteLine("请输入要发送的消息内容&＃xff1a;");
string msg &＃61; null;
while (!string.IsNullOrEmpty(msg &＃61; Console.ReadLine()))

var body &＃61; Encoding.UTF8.GetBytes(msg);
//body是byte类型&＃xff0c;使用了一个名为“”的默认交换机
channel.BasicPublish(exchange: "", routingKey: "hello", basicProperties: null, body: body);
Console.WriteLine("[x] Sent 0", msg);




Console.WriteLine("Press [Enter] to exit");
Console.Read();



当开启多个消费者之后&＃xff0c;默认会轮询消费。

3.2 ※手动发送ACK与数据持久存储

当消费者消费完成一个消息之后&＃xff0c;手动发送一条ack命令给broker。解决consumer突然死掉之后&＃xff0c;导致消息丢失的问题。如果mq一直没收到ack&＃xff0c;则会将此消息重新入队列&＃xff0c;给其他消费者进行消费。

生产者示例&＃xff1a;

• 将task_queue的durable设置为true&＃xff0c;这样及时Broker重启&＃xff0c;此队列也不会消失。
• 将消息的Persistent也设置为true。即消息也持久存储&＃xff0c;但是并不代表消息会100%不丢失&＃xff0c;它只是告诉MQ将消息存储在硬盘上。在MQ收到消息且写入硬盘之前如果挂了&＃xff0c;那消息就丢了。如果需要保证100%的可用&＃xff0c;可以使用后面小节的“发布确认”功能。

static void Main(string[] args)

var factory &＃61; new ConnectionFactory() HostName &＃61; "localhost" ;
using (var connection &＃61; factory.CreateConnection())

using (var channel &＃61; connection.CreateModel())

//durable设置为了true&＃xff0c;队列持久化
channel.QueueDeclare(queue: "task_quene", durable: true, exclusive: false, autoDelete: false, arguments: null);

var props &＃61; channel.CreateBasicProperties();
//消息持久化
props.Persistent &＃61; true;
Console.WriteLine("请输入要发送的消息内容&＃xff1a;");
string msg &＃61; null;
while (!string.IsNullOrEmpty(msg &＃61; Console.ReadLine()))

var body &＃61; Encoding.UTF8.GetBytes(msg);
channel.BasicPublish(exchange: "", routingKey: "task_quene", basicProperties: props, body: body);
Console.WriteLine("[x] Sent 0", msg);



Console.WriteLine("Press [Enter] to exit");
Console.Read();



消费者示例&＃xff1a;

默认情况下MQ会按照worker的顺序把队列里的消息一个个的分给worker&＃xff0c;这种分配消息的方式有一定的弊端&＃xff0c;假如有两个worker且队列里的消息根据耗时长短间隔排列。这样所有耗时长的消息都会被分给worker1&＃xff0c;短的分配给worker2. 造成worker2长时间空闲。所以就可以通过设置Qos的方式来改善&＃xff0c;channel.BasicQos(0, 1, false)表示broker一次只把1个消息发给worker&＃xff0c;直到这个worker发出了ack&＃xff0c;才继续把下一个消息分给他。

static void Main(string[] args)

var factory &＃61; new ConnectionFactory() HostName &＃61; "localhost" ;
using (var connection &＃61; factory.CreateConnection())

using (var channel &＃61; connection.CreateModel())

channel.QueueDeclare("task_quene", true, false, false, null);
//设置qos
channel.BasicQos(0, 1, false);
Console.WriteLine(" [*] Waiting for messages.");
var consumer &＃61; new EventingBasicConsumer(channel);
consumer.Received &＃43;&＃61;async (model, args) &＃61;>

byte[] body &＃61; args.Body.ToArray();
var msg &＃61; Encoding.UTF8.GetString(body);
Console.WriteLine($"[-] Task msg received");
await Task.Delay(msg.Length * 1000);//模拟耗时任务
Console.WriteLine(" [x] Task 0 Done", msg);
//手动发送ack,必须在同一个channel里发送
channel.BasicAck(args.DeliveryTag, false);
;
channel.BasicConsume("task_quene", false, consumer);
Console.WriteLine("Press [Enter] to exit");
Console.ReadKey();





3.3 消费模式

消费者消费消息有两种模式&＃xff1a;

1. 推&＃xff08;push&＃xff09;&＃xff1a;服务端主动推送消息到channel里&＃xff0c;然后消费者消费信道里的消息
2. 拉&＃xff08;pull&＃xff09;&＃xff1a;消费者手动从服务端拉去消息

在上面的例子中我们看到的其实就是推模式&＃xff0c;使用的是channel.BasicConsume方法。而拉模式需要使用channel.BasicGet方法。如&＃xff1a;

var response&＃61;channel.BasicGet("task_quene",autoAck:false);
var body&＃61;response.Body;
channel.BasicAck(response.Envelope.DeliveryTag,false);


注意&＃xff1a;
BasicGet一次只能获取一条消息&＃xff0c;且不能将其放到一个循环里来替代BasicComsume&＃xff0c;否则会严重影响RabbitMQ的性能。如果要实现高吞吐量&＃xff0c;则应该使用BasicConsume。

一个消息可以被多个消费者消费&＃xff0c;此时就用到了交换机。

4.1 交换机&＃xff08;Exchange&＃xff09;

回顾下我们之前的例子&＃xff1a;

• 一个生产者用来发送消息
• 一个队列用来缓存和存储这些消息
• 一个消费者用来接收消息

RabbitMQ消息模型的设计核心思想是&＃xff1a;生产者从来不把消息直接丢给队列&＃xff0c;它甚至都不知道要把消息丢给哪个队列。
取而代之的是生产者只需要把消息丢给交换机&＃xff08;exchange&＃xff09;。交换机决定把消息丢给哪个队列&＃xff0c;或者丢给哪些队列&＃xff0c;或者丢弃这个消息。

下图就是发布订阅的模型&＃xff1a;

交换机分为以下几种类型&＃xff1a;

• direct&＃xff1a;把消息路由到与RoutingKey完全匹配的队列中
• topic&＃xff1a;把消息路由到符合RoutingKey匹配规则的队列中
• headers&＃xff1a;不依赖路由键匹配规则路由消息。是根据发送消息内容中的headers属性进行匹配。性能差&＃xff0c;基本用不到。
• fanout&＃xff1a;把所有发送到该交换机的消息路由到所有与该交换机绑定的队列中

4.2 fanout交换机

1. 声明一个临时队列和一个交换机

消费者需要声明一个临时队列&＃xff0c;这个临时队列只能是消费者声明。当消费者断开连接时&＃xff0c;这个队列将会被删除。&＃xff08;此场景适用于我们的logs接收测试&＃xff0c;因为消费者不关系之前的日志是什么&＃xff09;。临时队列的名称类似于amq.gen-JzTY20BRgKO-HjmUJj0wLg格式。

消费者或者生产者也要声明一个交换机。

//创建临时队列&＃xff08;只能是消费者&＃xff09;
var quenuName &＃61; channel.QueueDeclare().QueueName;
//创建交换机&＃xff08;生产者或消费者&＃xff09;
channel.ExchangeDeclare("logs", ExchangeType.Fanout);


2. 将交换机与队列绑定

消费者需要将临时队列与交换机进行绑定。

channel.QueueBind(queue:quenuName, exchange:"logs",routingKey:"");


3. 最终模型与代码

与之前例子最大的不同是&＃xff0c;此时生产者需要把消息发送到交换机而不是某个队列上。在发送时我们就需要提供一个routingKey,但是fanout模式的交换机会忽略这个参数。

这样当我们发送消息时&＃xff0c;与exchange关联的所有队列都可以收到这个消息。

生产者&＃xff1a;

static void Main(string[] args)

var factory &＃61; new ConnectionFactory() HostName &＃61; "localhost" ;
using (var connection &＃61; factory.CreateConnection())

using (var channel &＃61; connection.CreateModel())

channel.ExchangeDeclare("logs", ExchangeType.Fanout);
Console.WriteLine("请输入要发送的消息内容&＃xff1a;");
string msg &＃61; null;
while (!string.IsNullOrEmpty(msg &＃61; Console.ReadLine()))

var body &＃61; Encoding.UTF8.GetBytes(msg);
channel.BasicPublish(exchange: "logs", routingKey: "", basicProperties: null, body: body);
Console.WriteLine("[x] Sent 0", msg);



Console.WriteLine("Press [Enter] to exit");
Console.Read();



消费者&＃xff1a;

这里我们使用channel.QueueDeclare().QueueName创建一个临时队列并返回队列名称。当消费者断开连接时&＃xff0c;这个队列将会被删除。&＃xff08;此场景试用与我们的logs接收&＃xff0c;因为消费者不关心之前的日志是什么&＃xff09;

static void Main(string[] args)

var factory &＃61; new ConnectionFactory() HostName &＃61; "localhost" ;
using (var connection &＃61; factory.CreateConnection())

using (var channel &＃61; connection.CreateModel())

channel.ExchangeDeclare("logs", ExchangeType.Fanout);
//创建一个临时队列
var queueName &＃61; channel.QueueDeclare().QueueName;
//交换机与队列的绑定
channel.QueueBind(queue:queueName, exchange:"logs",routingKey: "");
Console.WriteLine(" [*] Waiting for messages.");
var consumer &＃61; new EventingBasicConsumer(channel);
consumer.Received &＃43;&＃61; (model, args) &＃61;>

byte[] body &＃61; args.Body.ToArray();
var msg &＃61; Encoding.UTF8.GetString(body);
Console.WriteLine($"[-] Task msg received");
//手动发送ack,必须在同一个channel里发送
channel.BasicAck(args.DeliveryTag, false);
;
channel.BasicConsume(queueName, false, consumer);
Console.WriteLine("Press [Enter] to exit");
Console.ReadKey();





4.3 路由&＃xff08;Routing&＃xff09;

在上面的打印log例子中&＃xff0c;所有的消费者都能收到同一个log消息。在这一节我们将通过路由的方式来订阅消息的子集。例如一个消费者只用来接收critical级别的消息&＃xff0c;而其他消费者接收所有消息。

在上一小节中&＃xff0c;消费者将队列与交换机绑定时用到了channel.QueueBind方法&＃xff0c;表示“这个队列对这个交换机发出消息很感兴趣&＃xff0c;愿意接收这些消息”。这个绑定方法还接收一个routingKey参数&＃xff0c;取决于不同的交换机类型&＃xff0c;这个参数有可能会被忽略&＃xff08;例如我们之前用到的fanout交换机&＃xff09;。

4.3.1 Direct交换机

所以这里我们将以direct类型的交换机为例&＃xff0c;如果绑定队列时设置的routingKey等于发送消息时设置的routingKey&＃xff0c;这个队列就可以收到消息。举例如下&＃xff1a;

direct类型的交换机X下绑定了两个队列Q1和Q2。Q1的routingKey是orange&＃xff0c;Q2的routingKey是black和green。所以发送时如果消息的routingKey设置为orange则Q1会接收到&＃xff0c;如果是black或green则Q2会接收到&＃xff0c;如果是其他的值&＃xff0c;则会被交换